/*
    kOTF Open Trace Format - a library for generating execution traces.
    Copyright 2012 Daniel K. O.

    kOTF is free software: you can redistribute it and/or modify
    it under the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    kOTF is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public License
    along with kOTF.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "StreamImpl.hpp"
#include <kotf/Session.hpp>
#include <kotf/Stream.hpp>

#include "TypesInternal.hpp"
#include "EventBuffer.hpp"
#include "SessionImpl.hpp"


namespace kotf {
    
    namespace {
        const char magic_array[sizeof(Event)] = "KOTF-1.0.0stream";
    }
    const char * StreamImpl::magic = magic_array;




    StreamImpl::StreamImpl(Stream& str, Session& sess, uint64_t i, const string& n) :
        iface(str),
        sessionImpl(*sess.pimpl),
        eventBuffer(sessionImpl.getEventBuffer(this)),
        id(i),
        name(n),
        active(false),
        totalEvents(0),
        maxEvents(std::numeric_limits<size_t>::max())
    {
        string fname = sessionImpl.getName() + "-" + name + ".stream";
        file.open(fname.c_str(), std::ios_base::out | std::ios_base::binary);
        file.sputn(magic, sizeof(magic_array));
    }

    StreamImpl::~StreamImpl()
    {
        delete eventBuffer;
    }

    uint64_t
    StreamImpl::getId() const
    {return id;}
    

    const string&
    StreamImpl::getName() const
    {return name;}


    Session&
    StreamImpl::getSession()
    {return sessionImpl.getSession();}


    void
    StreamImpl::flush()
    {
        assert(active == false);
        sessionImpl.submit(eventBuffer);
        eventBuffer = sessionImpl.getEventBuffer(this);
    }

    void
    StreamImpl::write(EventBuffer *eb)
    {
        file.sputn(reinterpret_cast<char*>(eb->data), eb->size * sizeof(Event));
    }


    void
    StreamImpl::add(const Event& e)
    {
        if (totalEvents++ < maxEvents)
            while (!eventBuffer->add(e)) {
                sessionImpl.submit(eventBuffer);
                eventBuffer = sessionImpl.getEventBuffer(this);
            }
    }



    const State&
    StreamImpl::findState(const string& stn)
    {
        auto i = stateCache.find(stn);
        if (i != stateCache.end())
            return i->second; // fast path
        else
            return stateCache[stn] = sessionImpl.getState(stn); // slow path
    }

    const Counter&
    StreamImpl::findCounter(const string& cn)
    {
        auto i = counterCache.find(cn);
        if (i != counterCache.end())
            return i->second; // fast path
        else
            return counterCache[cn] = sessionImpl.getCounter(cn); // slow path
    }

    const Channel&
    StreamImpl::findChannel(const string& chn)
    {
        auto i = channelCache.find(chn);
        if (i !=channelCache.end())
            return i->second; // fast path
        else
            return channelCache[chn] = sessionImpl.getChannel(chn); // slow path
    }
    


    void
    StreamImpl::setLimit(size_t limit)
    {maxEvents = limit;}
    
    size_t
    StreamImpl::getLimit() const
    {return maxEvents;}
    



    bool
    StreamImpl::isActive() const
    {return active;}
    


    void
    StreamImpl::start(uint64_t tag)
    {
        assert(active == false);
        Event e = {{{0}}};
        e.stream.timestamp = sessionImpl.getTime();
        e.stream.type = Event::StreamBegin;
        e.stream.id = id;
        e.stream.parent = InvalidId;
        e.stream.tag = tag;
        add(e);
        active = true;
    }

    void
    StreamImpl::start(const Stream& parent, uint64_t tag)
    {
        assert(active == false);
        Event e = {{{0}}};
        e.stream.timestamp = sessionImpl.getTime();
        e.stream.type = Event::StreamBegin;
        e.stream.id = id;
        e.stream.parent = parent.getId();
        e.stream.tag = tag;
        add(e);
        active = true;
    }

    void
    StreamImpl::finish(uint64_t tag)
    {
        assert(active == true);
        Event e = {{{0}}};
        e.stream.timestamp = sessionImpl.getTime();
        e.stream.type = Event::StreamEnd;
        e.stream.id = id;
        e.stream.parent = InvalidId; // ignored, but let's initialize it anyway
        e.stream.tag = tag;
        add(e);
        active = false;
    }


    void
    StreamImpl::push(const State& st, uint64_t tag)
    {
        Event e = {{{0}}};
        e.state.timestamp = sessionImpl.getTime();
        e.state.type = Event::EnterState;
        e.state.id = st.id;
        e.state.tag = tag;
        add(e);
    }

    void
    StreamImpl::push(const string& stn, uint64_t tag)
    {
        Event e = {{{0}}};
        e.state.timestamp = sessionImpl.getTime();
        e.state.type = Event::EnterState;
        e.state.id = findState(stn).id;
        e.state.tag = tag;
        add(e);
    }

    void
    StreamImpl::pop(const State& st, uint64_t tag)
    {
        Event e = {{{0}}};
        e.state.timestamp = sessionImpl.getTime();
        e.state.type = Event::LeaveState;
        e.state.id = st.id;
        e.state.tag = tag;
        add(e);
    }

    void
    StreamImpl::pop(const string& stn, uint64_t tag)
    {
        Event e = {{{0}}};
        e.state.timestamp = sessionImpl.getTime();
        e.state.type = Event::LeaveState;
        e.state.id = findState(stn).id;
        e.state.tag = tag;
        add(e);
    }

    void
    StreamImpl::change(const State& oldst, const State& newst, uint64_t tag)
    {
        // pop and push, but with the same timestamp
        Event e = {{{0}}};
        e.state.timestamp = sessionImpl.getTime();
        e.state.type = Event::EnterState;
        e.state.id = oldst.id;
        e.state.tag = tag;
        add(e);
        e.state.type = Event::LeaveState;
        e.state.id = newst.id;
        add(e);
    }

    void
    StreamImpl::change(const string& oldstn, const string& newstn, uint64_t tag)
    {
        // pop and push, but with the same timestamp
        Event e = {{{0}}};
        e.state.timestamp = sessionImpl.getTime();
        e.state.type = Event::EnterState;
        const State& oldst = findState(oldstn);
        const State& newst = findState(newstn);
        e.state.id = oldst.id;
        e.state.tag = tag;
        add(e);
        e.state.type = Event::LeaveState;
        e.state.id = newst.id;
        add(e);
    }


    void
    StreamImpl::update(const Counter& c, uint64_t v, uint64_t tag)
    {
        Event e = {{{0}}};
        e.counter.timestamp = sessionImpl.getTime();
        e.counter.type = Event::UpdateCounter;
        e.counter.id = c.id;
        e.counter.value = v;
        e.counter.tag = tag;
        add(e);
    }

    void
    StreamImpl::update(const string& cn, uint64_t v, uint64_t tag)
    {
        Event e = {{{0}}};
        e.counter.timestamp = sessionImpl.getTime();
        const Counter& c = findCounter(cn);
        e.counter.type = Event::UpdateCounter;
        e.counter.id = c.id;
        e.counter.value = v;
        e.counter.tag = tag;
        add(e);
    }


    void
    StreamImpl::send(const Channel& ch, uint64_t length, uint64_t tag)
    {
        Event e = {{{0}}};
        e.message.timestamp = sessionImpl.getTime();
        e.message.type = Event::SendMsg;
        e.message.chan = ch.id;
        e.message.len = length;
        e.message.tag = tag;
        add(e);
    }

    void
    StreamImpl::send(const string& chn, uint64_t length, uint64_t tag)
    {
        Event e = {{{0}}};
        e.message.timestamp = sessionImpl.getTime();
        const Channel& ch = findChannel(chn);
        e.message.type = Event::SendMsg;
        e.message.chan = ch.id;
        e.message.len = length;
        e.message.tag = tag;
        add(e);
    }

    void
    StreamImpl::broadcast(const Channel& ch, uint64_t length, uint64_t tag)
    {
        Event e = {{{0}}};
        e.message.timestamp = sessionImpl.getTime();
        e.message.type = Event::BroadcastMsg;
        e.message.chan = ch.id;
        e.message.len = length;
        e.message.tag = tag;
        add(e);
    }
    
    void
    StreamImpl::broadcast(const string& chn, uint64_t length, uint64_t tag)
    {
        Event e = {{{0}}};
        e.message.timestamp = sessionImpl.getTime();
        const Channel& ch = findChannel(chn);
        e.message.type = Event::BroadcastMsg;
        e.message.chan = ch.id;
        e.message.len = length;
        e.message.tag = tag;
        add(e);
    }

    void
    StreamImpl::recv(const Channel& ch, uint64_t length, uint64_t tag)
    {
        Event e = {{{0}}};
        e.message.timestamp = sessionImpl.getTime();
        e.message.type = Event::RecvMsg;
        e.message.chan = ch.id;
        e.message.len = length;
        e.message.tag = tag;
        add(e);
    }

    void
    StreamImpl::recv(const string& chn, uint64_t length, uint64_t tag)
    {
        Event e = {{{0}}};
        e.message.timestamp = sessionImpl.getTime();
        const Channel& ch = findChannel(chn);
        e.message.type = Event::RecvMsg;
        e.message.chan = ch.id;
        e.message.len = length;
        e.message.tag = tag;
        add(e);
    }

}
